package com.dmp.storm.bolt.kafka;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.*;

public class KafkaProducer {
	private Producer<String, byte[]> inner;

	public KafkaProducer(Properties properties) throws Exception {
		//properties.put("zk.connect", properties.get("zookeeper.connect"));  
		//properties.put("serializer.class", "kafka.serializer.StringEncoder");  
		//properties.put("metadata.broker.list", Global.getConfig("metadata.broker.list")); 
		//properties.put("partitioner.class", "com.dmp.storm.bolt.kafka.KafkaPartitioner");
		//ProducerConfig config = new ProducerConfig(properties);
		//inner = new Producer<String, String>(config);
		Properties propertiesNew = new Properties();
		propertiesNew.put("zk.connect", properties.get("zookeeper.connect"));  
		//propertiesNew.put("serializer.class", "kafka.serializer.StringEncoder"); 
		propertiesNew.put("serializer.class", "kafka.serializer.DefaultEncoder");
		propertiesNew.put("key.serializer.class", "kafka.serializer.StringEncoder");
		propertiesNew.put("metadata.broker.list", properties.get("metadata.broker.list")); 
		propertiesNew.put("partitioner.class", "com.dmp.storm.bolt.kafka.KafkaPartitioner");
		ProducerConfig kafkaConfig = new ProducerConfig(propertiesNew);
		inner = new Producer<String, byte[]>(kafkaConfig);
	}

	public void send(String topicName, String message) {
		if (topicName == null || message == null) {
			return;
		}
		KeyedMessage<String, byte[]> km = new KeyedMessage<String, byte[]>(topicName, message.getBytes());
		inner.send(km);
	}
	
	public void send(String topicName, String key ,String message) {
		if (topicName == null || message == null) {
			return;
		}
		KeyedMessage<String, byte[]> km = new KeyedMessage<String, byte[]>(topicName,key,message.getBytes());
		inner.send(km);
	}

	public void send(String topicName, Collection<String> messages) {
		if (topicName == null || messages == null) {
			return;
		}
		if (messages.isEmpty()) {
			return;
		}
		List<KeyedMessage<String, byte[]>> kms = new ArrayList<KeyedMessage<String, byte[]>>();
		for (String entry : messages) {
			KeyedMessage<String, byte[]> km = new KeyedMessage<String, byte[]>(topicName, entry.getBytes());
			kms.add(km);
		}
		inner.send(kms);
	}
	
	
	public void batchSend(String topicName, List<Object> messages) {
		//java -classpath ****.jar ****.****.className [args]
		if (topicName == null || messages == null) {
			return;
		}
		if (messages.isEmpty()) {
			return;
		}
		List<KeyedMessage<String, byte[]>> kms = new ArrayList<KeyedMessage<String, byte[]>>();
		for (Object entry : messages) {
			Random r = new Random();
			KeyedMessage<String, byte[]> km = new KeyedMessage<String, byte[]>(topicName,""+r.nextInt(),entry.toString().getBytes());
			kms.add(km);
		}
		inner.send(kms);
	}

	public void close() {
		inner.close();
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		String[] urls = {"www.baidu.com","www.sina.com","www.qq.com","www.letv.com","www.360.com"};
		KafkaProducer producer = null;
		Properties properties = new Properties();
		properties.put("zk.connect", "tb-infopush7-ossweb:2181,tb-infopush9-ossweb:2181,tb-infopush8-ossweb:2181");
		properties.put("zookeeper.connect", "tb-infopush7-ossweb:2181,tb-infopush9-ossweb:2181,tb-infopush8-ossweb:2181");  
		properties.put("serializer.class", "kafka.serializer.StringEncoder");  
		properties.put("metadata.broker.list", "tb-infopush7-ossweb:6667,tb-infopush9-ossweb:6667,tb-infopush8-ossweb:6667"); 
		properties.put("partitioner.class", "com.dmp.storm.bolt.kafka.KafkaPartitioner");
		
		String str1="12,12,11,123,123,20170314194145,1,1,20170314194145,1,13,1,1";
		String str2="1,11,11,11,11,20170314194145,1,1,20170314194145,1,12,1,1";
		String str3="12,1,11,11,11,20170314194145,11,11,20170314194145,11,12,3,1";
		String str4="20170314194145,20170314194145,112,1,11,11,1,1,1,1,1,1,1";
		String str5="20170314194145,1,1,1,1,1,1,1";
		String str6="1,20170314194145,1,1,1,1,1,1,1,1,1,1,1,1,2,1,1,1,1";
		String str7="20170314194145,20170314194145,1,1,2,1,2,1,2,2,2";
		String str8="20170314194145,20170314194145,12,1,22,1,1,12,12,1,1,2,12,1,2";
		
		
		try {
			 String bootMode="1";
			 if (args.length > 0) {
		        bootMode = args[0];
		        
		      }
			 producer = new KafkaProducer(properties);
			 Random r = new Random();
			 int i = 0;
			 if(bootMode.equals("1")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str1.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_click", data.toString());
			 }
			 if(bootMode.equals("2")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str2.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_close", data.toString());
			 }
			 if(bootMode.equals("3")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str3.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_expose", data.toString());
			 }
			 if(bootMode.equals("4")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str4.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_fail", data.toString());
			 }
			 if(bootMode.equals("5")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str5.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_fe", data.toString());
			 }
			 if(bootMode.equals("6")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str6.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_get", data.toString());
			 }
			 if(bootMode.equals("7")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str7.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_input", data.toString());
			 }
			 if(bootMode.equals("8")){
				 StringBuffer data = new StringBuffer();
				 String[] values = str8.split(",");
				 for(int j=0; i<values.length; j++){
					 if(j==0){
						 data.append(values[j]);
					 }else{
						 data.append("\\001");
						 data.append(values[j]);
					 }
				 }
				 System.out.println(data.toString());
				 producer.send("log_detail_success", data.toString());
			 }
			
			/*while (true) {
				long time = System.currentTimeMillis();
				String serialNumber = "08980000000135";
				String keywords = "";
				String url = urls[r.nextInt(5)];
				StringBuffer data = new StringBuffer();
				data.append(time)
				.append("\\001")
				.append(serialNumber)
				.append("\\001")
				.append(keywords)
				.append("\\001")
				.append(url);
				System.out.println(data.toString());
				producer.send("ad-topic5", data.toString());
				i++;
				Thread.sleep(500000);
			}*/
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (producer != null) {
				producer.close();
			}
		}

	}
}
